package com.github.ghsea.scheduler.core;

import com.github.ghsea.scheduler.dto.Location;
import com.github.ghsea.scheduler.dto.TaskExecutionEvent;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @author guhai
 * @since 2021/6/17 21:47
 */

@Component
@Slf4j
public class TaskExecutor {

    public void produceTaskEvent(TaskExecutionEvent task) {
        log.info("produce event:{}" + task);

        //TODO 空实现
        sendTaskEventToKafka(task);
    }

    public void executeTask(List<TaskExecutionEvent> tasks) {
        //TODO 消费Kafka消息
        tasks.forEach(e -> log.info("execute event:{}" + e));

        List<String> plateNumbers = tasks.stream().map(e -> e.getBizCode()).collect(Collectors.toList());

        List<Location> locations = getCurrentLocations(plateNumbers);

        saveHbase(locations);

        sendLocationToKafka(locations);
    }

    /**
     * 通过http从三方拉取最新轨迹
     *
     * @param plateNumbers
     * @return
     */
    private List<Location> getCurrentLocations(List<String> plateNumbers) {
        return new ArrayList<>();
    }

    /**
     * 将轨迹点存入Hbase待后面查询车辆轨迹用
     *
     * @param locations
     */
    private void saveHbase(List<Location> locations) {

    }

    private void sendLocationToKafka(List<Location> locations) {

    }

    private void sendTaskEventToKafka(TaskExecutionEvent task){

    }

}
